package com.atguigu.userprofile.app

import com.atguigu.userprofile.common.bean.TagInfo
import com.atguigu.userprofile.common.constant.ConstCode
import com.atguigu.userprofile.common.dao.TagInfoDAO
import com.atguigu.userprofile.common.util.ClickhouseUtil
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import scala.collection.mutable.ListBuffer

object TaskBitmap {

  // 宽表写入Bitmap表
  // 拼接一条 转储的 insert select SQL 在clickhouse中执行
  //  要把标签按照标签值类型，分成4份  ，形成4个sql，分别插入到不同的表中
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("task_bitmap_app").setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()


    val busiDate: String = args(1)

    //1 取得全体标签
    val tagInfoListAll: List[TagInfo] = TagInfoDAO.getTagInfoList()

    //2   要把标签按照标签值类型，分成4份
    val tagInfoListString: ListBuffer[TagInfo] = ListBuffer()
    val tagInfoListLong: ListBuffer[TagInfo] = ListBuffer()
    val tagInfoListDecimal: ListBuffer[TagInfo] = ListBuffer()
    val tagInfoListDate: ListBuffer[TagInfo] = ListBuffer()

    for (tagInfo <- tagInfoListAll) {
      if (tagInfo.tagValueType.equals(ConstCode.TAG_VALUE_TYPE_STRING)) {
        tagInfoListString.append(tagInfo)
      } else if (tagInfo.tagValueType.equals(ConstCode.TAG_VALUE_TYPE_DECIMAL)) {
        tagInfoListDecimal.append(tagInfo)
      } else if (tagInfo.tagValueType.equals(ConstCode.TAG_VALUE_TYPE_LONG)) {
        tagInfoListLong.append(tagInfo)
      } else if (tagInfo.tagValueType.equals(ConstCode.TAG_VALUE_TYPE_DATE)) {
        tagInfoListDate.append(tagInfo)
      }
    }


    // 3  形成4个sql，分别插入到不同的表中
    insertTagsToBitmap(tagInfoListString, "user_tag_value_string", busiDate)
    insertTagsToBitmap(tagInfoListDecimal, "user_tag_value_decimal", busiDate)
    insertTagsToBitmap(tagInfoListLong, "user_tag_value_long", busiDate)
    insertTagsToBitmap(tagInfoListDate, "user_tag_value_date", busiDate)


  }

  // 3  形成4个sql，分别插入到不同的表中
  def insertTagsToBitmap(tagInfoList: ListBuffer[TagInfo], toTableName: String, busiDate: String): Unit = {


    //insert into xxxx
    //select  tg.1 tag_code,tg.2 tag_value ,groupBitmapState( cast(uid as UInt64)  )  us ,'2020-06-14'  from
    //(
    //     select  uid,  arrayJoin([('tg_person_base_gender',tg_person_base_gender) ,
    //       ('tg_person_base_agegroup',tg_person_base_agegroup)])  tg
    //	   from up_tag_merge_20200614
    //)utg
    //group by tg.1 ,tg.2
    if (tagInfoList != null && tagInfoList.size > 0) {
      var tagCodeSQL=""
      if(toTableName.equals("user_tag_value_decimal")||toTableName.equals("user_tag_value_long")){
        tagCodeSQL= tagInfoList.map(tagInfo => s"('${tagInfo.tagCode.toLowerCase}', if(${tagInfo.tagCode.toLowerCase}='' ,'0',${tagInfo.tagCode.toLowerCase}))").mkString(",")

      }else {
        tagCodeSQL= tagInfoList.map(tagInfo => s"('${tagInfo.tagCode.toLowerCase}', ${tagInfo.tagCode.toLowerCase} )").mkString(",")

      }




      val fromTable = s"up_tag_merge_${busiDate.replace("-", "")}"

      //数据清理的范围  清理分区
      val clearSQL=s" alter table $toTableName delete where dt='$busiDate'"
      println(clearSQL)

      ClickhouseUtil.executeSql(clearSQL)
      val insertSQL =
        s"""
           |     insert into $toTableName
           |     select  tg.1 tag_code,tg.2 tag_value ,groupBitmapState( cast(uid as UInt64)  )  us ,'$busiDate'  from
           |     (
           |          select  uid,  arrayJoin([$tagCodeSQL ])  tg
           |    	   from $fromTable
           |     )utg
           |     group by tg.1 ,tg.2
           |""".stripMargin
      println(insertSQL)


      ClickhouseUtil.executeSql(insertSQL)
    }

  }


}
